package hotnet.nio;

import java.io.IOException;
import java.net.ServerSocket;
import java.net.SocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Collection;
import java.util.Iterator;

import hotnet.config.DefaultIoSocketSessionConfig;
import hotnet.config.IoSessionConfig;
import hotnet.processor.IoProcessor;
import hotnet.service.AbstractPollingIoAcceptor;
import hotnet.service.IoService;
import hotnet.service.SocketAcceptor;
import hotnet.session.IoSession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class NioSocketAcceptor extends AbstractPollingIoAcceptor implements SocketAcceptor {
	private Selector selector;
	private Logger logger = LoggerFactory.getLogger(NioSocketAcceptor.class);
	
	public NioSocketAcceptor() {
		super(new DefaultIoSocketSessionConfig(), NioSocketProcessor.class);
	}
	
	public NioSocketAcceptor(IoSessionConfig config, Class <? extends IoProcessor> processorCls) {
		super(config, processorCls);
	}
	
	@Override
	public IoSession newSession(SocketAddress remoteAddress,
			SocketAddress localAddress) {
		return null;
	}

	@Override
	protected void init() throws Exception {
		selector =  Selector.open();
	}
	
	@Override
	protected IoSession accept(IoProcessor processor, ServerSocketChannel handle)
			throws Exception {
		
		SelectionKey key = null;
		
		if (handle != null) {
			key = handle.keyFor(selector);
		}
		
		if (key != null && !(key.isValid()) && (!key.isAcceptable())) {
			return null;
		}
		
		SocketChannel ch = handle.accept();
		
		IoSession session = new NioSocketSession(processor, ch, this);
		return session;
	}

	@Override
	protected void close(ServerSocketChannel handle) throws Exception {
		SelectionKey key = handle.keyFor(selector);
		
		if (key != null) {
			key.cancel();
		}
		
		handle.close();
	}

	@Override
	protected int select() throws Exception {
		return selector.select();
	}

	@Override
	protected void wakeupAcceptor() {
		selector.wakeup();
		
	}

	@Override
	protected ServerSocketChannel open(SocketAddress localAddress)
			throws Exception {
		ServerSocketChannel channel = null;
		
		channel = ServerSocketChannel.open();
		
		boolean success = false;
		
		try {
			channel.configureBlocking(false);
			ServerSocket socket = channel.socket();
			
			socket.setReuseAddress(isReuseAddress());
			
			try {
				socket.bind(localAddress);
			} catch (IOException e) {
				channel.close();
				throw e;
			}
			
			channel.register(selector, SelectionKey.OP_ACCEPT);
			success = true;
			
		} finally {
			if (!success) {
				channel.close();
			}
		}
		
		return channel;
	}

	@Override
	protected Iterator<ServerSocketChannel> selectedHandles() {
		return new ServerSocketChannelIterator(selector.selectedKeys());
	}
	
	public static class ServerSocketChannelIterator implements Iterator<ServerSocketChannel> {
		private final Iterator<SelectionKey> iterator;
		
		private ServerSocketChannelIterator(Collection<SelectionKey> selectedKeys) {
            iterator = selectedKeys.iterator();
        }
		
		@Override
		public boolean hasNext() {
			return iterator.hasNext();
		}

		@Override
		public ServerSocketChannel next() {
            SelectionKey key = iterator.next();

            if (key.isValid() && key.isAcceptable()) {
                return (ServerSocketChannel) key.channel();
            }

            return null;
		}

		@Override
		public void remove() {
			iterator.remove();
			
		}
		
	}

	@Override
	public IoService getService() {
		return this;
	}
}
